package com.panlm.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;

/**
 * 参数优先级排序：
 * （1）客户端代码中设置的值 >（2）ClassPath 下的用户自定义配置文件 >
 * （3）然后是服务器的自定义配置（xxx-site.xml）>（4）服务器的默认配置（xxx-default.xml）
 */
public class HdfsClient {

    private FileSystem fs;

    // 获取文件系统
    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        URI uri = new URI("hdfs://hadoop102:8020");
        Configuration conf = new Configuration();
        conf.set("dfs.replication", "2");
        fs = FileSystem.get(uri, conf, "panlm");
    }

    // 关闭资源
    @After
    public void close() throws IOException {
        fs.close();
    }

    // 创建目录
    @Test
    public void testMkdir() throws IOException {
        fs.mkdirs(new Path("hdfs://hadoop102/xiyou/huaguoshan"));
    }

    // 上传文件
    @Test
    public void testCopyFromLocalFile() throws IOException {
        /**
         * boolean delSrc: 是否删除源文件
         * boolean overwrite: 是否覆盖
         * Path src: 源文件路径
         * Path dst: 上传路径
        */
        fs.copyFromLocalFile(false, true,
                new Path("E:\\Hadoop\\sunwukong2.txt"),
                new Path("hdfs://hadoop102/xiyou/huaguoshan"));
    }

    // 下载文件
    @Test
    public void testCopyToLocalFile() throws IOException {
        /**
         * boolean delSrc: 是否删除源文件
         * Path src: 源文件路径
         * Path dst: 下载到本地的路径
         * boolean useRawLocalFileSystem: 是否使用RawLocalFileSystem作为本地文件系统
         */
        fs.copyToLocalFile(true,
                new Path("hdfs://hadoop102/xiyou/huaguoshan/sunwukong.txt"),
                new Path("E:\\Hadoop"), true);
    }

    //文件更名和移动
    @Test
    public void testRename() throws IOException {
        //修改文件名
        //fs.rename(new Path("/input/word.txt"), new Path("/input/wordcount.txt"));

        //移动文件并修改文件
        //fs.rename(new Path("hdfs://hadoop102/input/wordcount.txt"),
        //        new Path("hdfs://hadoop102/word.txt"));

        //修改文件名
        fs.rename(new Path("/input"), new Path("/output"));
    }

    //删除文件和目录
    @Test
    public void testDeleteFiles() throws IOException {
        //删除文件
        //fs.delete(new Path("/xiyou/huaguoshan/sunwukong2.txt"), true);

        //删除目录
        fs.delete(new Path("/xiyou/huaguoshan"), true);
    }

    //文件详情查看
    @Test
    public void testListFiles() throws IOException {
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        while (listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();

            System.out.println("======" + fileStatus.getPath() + "======");
            System.out.println(fileStatus.getPermission());
            System.out.println(fileStatus.getOwner());
            System.out.println(fileStatus.getGroup());
            System.out.println(fileStatus.getLen());
            System.out.println(fileStatus.getModificationTime());
            System.out.println(fileStatus.getReplication());
            System.out.println(fileStatus.getBlockSize());
            System.out.println(fileStatus.getPath().getName());

            //获取块信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            System.out.println(Arrays.toString(blockLocations));
        }
    }

    //文件和文件夹判断
    @Test
    public void testListFilesStatus() throws IOException {
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            //如果是文件
            if (fileStatus.isFile()) {
                System.out.println("f:" + fileStatus.getPath().getName());
            } else {
                System.out.println("d:" + fileStatus.getPath().getName());
            }
        }
    }
}
